package m3;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.Scanner;

public class Producer {

    public static void main(String[] args) throws MQClientException {

        // 创建事务消息生产者实例
        TransactionMQProducer t = new TransactionMQProducer("producerGroup3");

        // 设置name server
        t.setNamesrvAddr("192.168.64.141:9876");
        // 设置事务消息监听器
        t.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                System.out.println("执行本地业务，业务参数： " + args);

                if (true) {
                    // 一般因为网络原因
                    return LocalTransactionState.UNKNOW;
                }

                if(Math.random() < 0.5) {
                    System.out.println("事务执行成功，向Rocketmq服务器提交消息");
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
                System.out.println("事务执行失败，从Rocketmq服务器撤回消息、回滚消息");
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                System.out.println("正在处理Rocketmq服务器的事务回查");
                return LocalTransactionState.UNKNOW;
            }
        });
        // 启动
        t.start();

        // 发送事务消息，会触发执行事务消息监听器来执行本地事务
        while (true) {
            System.out.println("输入消息： ");
            String s = new Scanner(System.in).nextLine();
            Message msg = new Message("Topic3", s.getBytes());
            t.sendMessageInTransaction(msg, "假的业务数据");
        }

    }

}
